package edu.wisc.hevea.broker;

import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Level;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.wisc.application.ApplicationPolicy;

import edu.wisc.hevea.constants.ApplicationConstants;
import edu.wisc.hevea.messaging.MessagingServer;
import edu.wisc.hevea.messaging.MessagingServiceClient;
import edu.wisc.hevea.topic.TopicManager;


public class MessageBroker implements Runnable {

	private static final Logger logger = LoggerFactory
		      .getLogger(MessageBroker.class);

	  
	private String myHostName;
	private ApplicationPolicy applicationPolicy;
	private boolean isLeader;
	
	private Set<String> allBrokers;
	private Set<String> aliveBrokers;
	private ConcurrentSkipListSet<String> replicationBrokers;
	private String leaderBroker;
	
	
	private LeaderElectionThread leaderElectionThread;
	private HeartBeatThread heartBeatThread;
	private MessagingServer messagingServer;
	private TopicManager topicManager;
	private AutoScalingThread autoScalingThread;
	
	
	private MessagingServiceClient messagingServiceClient;
	

	public MessageBroker(String myHostName, ApplicationPolicy applicationPolicy) {
		this.setMyHostName(myHostName);
		this.setApplicationPolicy(applicationPolicy);
		this.replicationBrokers = new ConcurrentSkipListSet<String>();		
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub

		org.apache.log4j.BasicConfigurator.configure();
		org.apache.log4j.Logger.getRootLogger().setLevel(Level.INFO);
		
		if (args.length < 1) {
			System.err
					.println(" Please enter the broker's hostname to connect!");
			return;
		}

		// Read application policy
		String appID = "APP_ID_1";
		short replication = 2;
		List<String> dcsHosted = Arrays.asList(new String[] {
				ApplicationConstants.VIRGINIA_DC, ApplicationConstants.IAD_DC,
				ApplicationConstants.DUBLIN_DC });
		ApplicationPolicy appPolicy = new ApplicationPolicy(appID, replication, dcsHosted);

		String hostName = args[0];
		MessageBroker broker = new MessageBroker(hostName, appPolicy);
		broker.setLeader(false);
		Thread brokerThread = new Thread(broker);
		brokerThread.start();

		try {
			Thread.currentThread().join();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	@Override
	public void run() {
		// TODO Auto-generated method stub
		ScheduledExecutorService service = Executors
				.newSingleThreadScheduledExecutor();
		
		logger.info(this.toString());
		leaderElectionThread = new LeaderElectionThread(this);
		heartBeatThread = new HeartBeatThread(this);
		setMessagingServiceClient(MessagingServiceClient.getMessagingServiceClient(this));
		topicManager = new TopicManager(this);
		messagingServer = new MessagingServer(this,topicManager);
		autoScalingThread = new AutoScalingThread(this);
		
		service.execute(leaderElectionThread);		
		service.execute(heartBeatThread);
		service.execute(messagingServer);
		
		service.scheduleWithFixedDelay(topicManager, 20 * 1000, 20 * 1000, TimeUnit.MILLISECONDS);
		service.scheduleWithFixedDelay(autoScalingThread, 30 * 1000, 10 * 1000, TimeUnit.MILLISECONDS);
		
		while(true) {
			try {
				Thread.sleep(60 * 1000 * 10);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				// Clean up code
				e.printStackTrace();
			}
		}
	}

	@Override
	public String toString() {
		return "HostName :" + getMyHostName() + " "
				+ getApplicationPolicy().toString();
	}
	
	public String getMyHostName() {
		return myHostName;
	}

	public void setMyHostName(String myHostName) {
		this.myHostName = myHostName;
	}

	public ApplicationPolicy getApplicationPolicy() {
		return applicationPolicy;
	}

	/**
	 * @return the leaderElectionThread
	 */
	public LeaderElectionThread getLeaderElectionThread() {
		return leaderElectionThread;
	}

	/**
	 * @param leaderElectionThread the leaderElectionThread to set
	 */
	public void setLeaderElectionThread(LeaderElectionThread leaderElectionThread) {
		this.leaderElectionThread = leaderElectionThread;
	}

	/**
	 * @return the heartBeatThread
	 */
	public HeartBeatThread getHeartBeatThread() {
		return heartBeatThread;
	}

	/**
	 * @param heartBeatThread the heartBeatThread to set
	 */
	public void setHeartBeatThread(HeartBeatThread heartBeatThread) {
		this.heartBeatThread = heartBeatThread;
	}

	public void setApplicationPolicy(ApplicationPolicy applicationPolicy) {
		this.applicationPolicy = applicationPolicy;
	}

	public boolean isLeader() {
		return isLeader;
	}

	public void setLeader(boolean isLeader) {
		this.isLeader = isLeader;
	}

	public Set<String> getAllBrokers() {
		return allBrokers;
	}

	public void setAllBrokers(Set<String> allBrokers) {
		this.allBrokers = allBrokers;
	}

	public Set<String> getAliveBrokers() {
		return aliveBrokers;
	}

	public void setAliveBrokers(Set<String> aliveBrokers) {
		this.aliveBrokers = aliveBrokers;
	}

	/**
	 * @return the messagingServer
	 */
	public MessagingServer getMessagingServer() {
		return messagingServer;
	}

	/**
	 * @param messagingServer the messagingServer to set
	 */
	public void setMessagingServer(MessagingServer messagingServer) {
		this.messagingServer = messagingServer;
	}

	public ConcurrentSkipListSet<String> getReplicationBrokers() {
		return replicationBrokers;
	}

	public void setReplicationBrokers(ConcurrentSkipListSet<String> replicationBrokers) {
		this.replicationBrokers = replicationBrokers;
	}

	public MessagingServiceClient getMessagingServiceClient() {
		return messagingServiceClient;
	}

	public void setMessagingServiceClient(MessagingServiceClient messagingServiceClient) {
		this.messagingServiceClient = messagingServiceClient;
	}

	public TopicManager getTopicManager() {
		return topicManager;
	}

	public void setTopicManager(TopicManager topicManager) {
		this.topicManager = topicManager;
	}

	public String getLeaderBroker() {
		return leaderBroker;
	}

	public void setLeaderBroker(String leaderBroker) {
		this.leaderBroker = leaderBroker;
	}

	public AutoScalingThread getAutoScalingThread() {
		return autoScalingThread;
	}

	public void setAutoScalingThread(AutoScalingThread autoScalingThread) {
		this.autoScalingThread = autoScalingThread;
	}
}
